#include "EventLoop.h"
#include <iostream>
// #include "Log/Logging.h"
#include <unistd.h>
#include <assert.h>
#include <signal.h>
#include <sys/eventfd.h>
#include "Channel.h"
#include "EPoller.h"
#include "TimerQueue.h"
#include "TcpConnection.h"

using namespace std::placeholders;

// __thread是GCC内置线程局部存储设施，__thread变量每个线程有一份独立的实体
__thread EventLoop* t_loopInThisThread = 0;

const int kPollTimeMs = -1;

static int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        // LOG_ERROR << "Failed in eventfd";
        abort();
    }
    return evtfd;
}

class IgnoreSigPipe {
public:
    IgnoreSigPipe() { ::signal(SIGPIPE, SIG_IGN); }
};
IgnoreSigPipe initObj;

EventLoop::EventLoop()
    : looping_(false),
      quit_(false),
      callingPendingFunctors_(false),
      threadId_(this_thread::get_id()),
      poller_(new EPoller(this)),
      timerQueue_(new TimerQueue(this)),
      wakeupFd_(createEventfd()),
      wakeupChannel_(new Channel(this, wakeupFd_)),
      connectionId_(0)
{

    if (t_loopInThisThread) {
        // LOG_FATAL << "Another eventLoop " << t_loopInThisThread << " exists in thread " << threadId_;
        // LOG_FATAL << "Another eventLoop ";
    } else {
        t_loopInThisThread = this;
    }
    wakeupChannel_->setReadCallback( std::bind(&EventLoop::handleRead, this) );
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop() {
    assert(!looping_);
    t_loopInThisThread = nullptr;
}

void EventLoop::abortNotInLoopThread()
{
    std::cout << "!!!!!!!!!!!!!!!! abortNotInLoopThread !!!!!!!!!!!!!!!\n";
}

void EventLoop::loop() {
    looping_ = true;
    quit_ = false;
    while (!quit_)
    {
        activeChannels_.clear();
        // 阻塞时长kPollTimeMs
        int numEvents = poller_->poll(kPollTimeMs, &activeChannels_);
        for (ChannelList::iterator it = activeChannels_.begin();
            it != activeChannels_.end(); ++it)
        {
            (*it)->handleEvent();
        }
        doPendingFunctors();
    }
    looping_ = false;
}

void EventLoop::quit()
{
    quit_ = true;
    if (!isInLoopThread())
    {
        wakeup();
    }
}

void EventLoop::updateChannel(Channel* channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    poller_->removeChannel(channel);
}

TimerId EventLoop::runAt(const Timestamp& time, const TimerCallback& cb) {
    return timerQueue_->addTimer(cb, time, 0.0);
}

TimerId EventLoop::runAfter(double delay, const TimerCallback& cb) {
    Timestamp time(addTime(Timestamp::now(), delay));
    return runAt(time, cb);
}

TimerId EventLoop::runEvery(double interval, const TimerCallback& cb) {
    Timestamp time(addTime(Timestamp::now(), interval));
    return timerQueue_->addTimer(cb, time, interval);
}

void EventLoop::runInLoop(const Functor& cb) {
    if (isInLoopThread()) {
        cb();
    }
    else {
        queueInLoop(cb);
    }
}

void EventLoop::queueInLoop(const Functor& cb)
{
    {
    std::lock_guard<std::mutex> lock(mutex_);
    pendingFunctors_.push_back(cb);
    }

    // 不在当前线程或callingPendingFunctors_为真
    if (!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup();
    }
}

void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof one);
    if (n != sizeof one) {
        // LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
    }
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof one);
    if (n != sizeof one) {
        // LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
    }
}

void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        lock_guard<mutex> lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for (size_t i = 0; i < functors.size(); ++i) {
        functors[i]();
    }
    callingPendingFunctors_ = false;
}

void EventLoop::newConnection(int sockfd, const ConnectionCallback& connectionCallback, const MessageCallback& messageCallback) {
	char buf[32];
	snprintf(buf, sizeof buf, "#%d", connectionId_);
	++connectionId_;
	std::string connName = buf;

	TcpConnectionPtr conn( new TcpConnection(this, connName, sockfd) );
	connections_[conn->name()] = conn;
	conn->setConnectionCallback(connectionCallback);
	conn->setMessageCallback(messageCallback);
    conn->setCloseCallback(std::bind(&EventLoop::removeInMap, this, _1));
    conn->connectEstablished();
}

void EventLoop::insertInMap(const TcpConnectionPtr& conn) {
    conn->setCloseCallback(std::bind(&EventLoop::removeInMap, this, _1));
    connections_[conn->name()] = conn;
    conn->connectEstablished();
}

void EventLoop::removeInMap(const TcpConnectionPtr& conn) {
    size_t n = connections_.erase(conn->name());
    if (n != 1) printf("remove in map error\n");
    conn->connectDestroyed();
}